package cn.xuexiyuan.flinkstudy.state;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
 * @Description: 使用 KeyStte 中的 ValueState 获取流数据中的最大值, 实际中可以直接使用 maxBy 即可
 *
 * @Author 左龙龙
 * @Date 21-3-26
 * @Version 1.0
 **/
public class StateDemo01_KeyedState {

    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        // 1.source
        DataStreamSource<Tuple2<String, Long>> tupleDs = env.fromElements(
                Tuple2.of("北京", 1L),
                Tuple2.of("上海", 2L),
                Tuple2.of("北京", 4L),
                Tuple2.of("北京", 6L),
                Tuple2.of("北京", 7L),
                Tuple2.of("上海", 1L),
                Tuple2.of("上海", 3L),
                Tuple2.of("上海", 1L)
        );

        // 2.transformation
        // 需求：各个城市最大的 value 值
        SingleOutputStreamOperator<Tuple2<String, Long>> result = tupleDs.keyBy(t -> t.f0).maxBy(1);

        // 学习时使用　KeyState 中的 ValueState 来实现 maxBy 的底层实现
        SingleOutputStreamOperator<Tuple3<String, Long, Long>> result2 = tupleDs.keyBy(t -> t.f0).map(new RichMapFunction<Tuple2<String, Long>, Tuple3<String, Long, Long>>() {
            // 1. 定义一个状态用来存放最大值
            private ValueState<Long> maxValueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                // 2. 创建状态描述器
                ValueStateDescriptor stateDescriptor = new ValueStateDescriptor("maxValueState", Long.class);
                // 3. 根据状态描述器获取初始化状态
                maxValueState = getRuntimeContext().getState(stateDescriptor);
            }


            @Override
            public Tuple3<String, Long, Long> map(Tuple2<String, Long> value) throws Exception {
                // 获取状态
                Long historyValue = maxValueState.value();
                if (null == historyValue || value.f1 > historyValue) {
                    // 更新状态
                    maxValueState.update(value.f1);
                    return Tuple3.of(value.f0, value.f1, value.f1);
                } else {
                    return Tuple3.of(value.f0, value.f1, historyValue);
                }
            }
        });

        // 3.sink
        result.print();
        result2.print("ValueState");

        // 4.excute
        env.execute();
    }

}
